package com.yjd.bi.cannal;

import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.yjd.bi.canal.enums.CanalDestinationEnums;
import com.yjd.bi.canal.model.CanalDeployerModel;
import com.yjd.bi.canal.service.ICanalDeployerService;
import com.yjd.comm.annotation.ServiceRemoteProxy;
import com.yjd.comm.base.control.BaseControl;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.rpm.RpmServiceKeyEnum;
import com.yjd.comm.util.DateUtil;
import com.yjd.comm.util.FrameUtil;
import com.yjd.comm.util.ServiceHolder;
import com.yjd.comm.util.ServletUtil;
import com.yjd.comm.util.StringUtil;
@Controller
@Scope("prototype")
@RequestMapping("/canal/")
public class ClientManageControl extends BaseControl{
	private static Map<String,CannelClient> clis;
	
	private static List<CanalDeployerModel> deployers;

	@ServiceRemoteProxy(serviceUrlResouceKey=RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	private ICanalDeployerService canalDeployerService;
	
	public static List<CanalDeployerModel> getDeployers() throws Exception {
		if(deployers==null){
			refreshDeployers();
		}
		return deployers;
	}
	
	public static void refreshDeployers() throws Exception{
		deployers=ServiceHolder.getService(ICanalDeployerService.class, "canalDeployerService",RpmServiceKeyEnum.RPM_BI_SERVICE_KEY_E).getModelList(FrameUtil.newHashMap(), DbWREnums.WRITE);
	}

	@RequestMapping("/info")
	public String getThreadInfo() throws NumberFormatException, Exception{
		init();
		request.setAttribute("info",getClientStatus());
		request.setAttribute("list", getDeployers());
		return "canal.info";
	}
	
	@RequestMapping("/stop")
	public void stop(String destination)throws Exception{
		CannelClient c=clis.get(destination);
		c.stop();
		this.successMsg("操作成功！");
	}
	
	@RequestMapping("/start")
	public void start(String destination)throws Exception{
		CannelClient c=clis.get(destination);
		c.start();
		this.successMsg("操作成功！");
	}
	
	public Map getClientStatus(){
		Map<String,String> status=FrameUtil.newHashMap();
		for(CannelClient c:clis.values()){
			status.put(c.destination, c.isWorking()?c.connector.checkValid()?"是":"connector无法连接":"否");
		}
		return status;
	}
	
	public synchronized void init() throws NumberFormatException, Exception{
		if(clis==null||clis.size()==0){
			clis=FrameUtil.newHashMap();
			for(CanalDeployerModel e:getDeployers()){
				 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(StringUtil.isEmpty(e.getIp_addr())?AddressUtils.getHostIp():e.getIp_addr(),
				            Integer.parseInt(e.getPort())), e.getDestination(), StringUtil.isEmpty(e.getUsername())?"":e.getUsername(), StringUtil.isEmpty(e.getPassword())?"":e.getUsername());
				 final CannelClient c=new CannelClient(e.getDestination(),connector) ;
				c.start();
				clis.put(e.getDestination(),c);
			}
		}
	}
	
	protected String buildPositionForDump(Entry entry) {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DateUtil.PATTERN_yyyy_MM_dd_HH_mm_ss);
        return entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
               + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
    }
	
	public synchronized void refresh() throws NumberFormatException, Exception{
			if(clis.size()!=CanalDestinationEnums.values().length){
				for(CannelClient c:clis.values()){
					c.stop();
				}
				clis.clear();;
				init();
			}
	}
	
	public void addClient(){
		
	}
}
